package com.xj.center.manager.service.kafka;

import cn.hutool.core.convert.Convert;
import com.xj.center.manager.service.BatchService;
import com.xj.center.manager.service.DriverSdkService;
import com.xj.common.bean.constant.Common;
import com.xj.common.bean.driver.DriverConfiguration;
import com.xj.common.bean.driver.DriverRegister;
import com.xj.common.model.DriverEvent;
import com.xj.common.utils.JsonPaseUtil;
import com.xj.common.utils.KafkaProducer;
import com.xj.common.utils.RedisUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
* 类说明  MANAGER层 驱动事件接收处理
*/
@Component
public class DriverEventConsumer {
    @Resource
    private KafkaProducer kafkaProducer;

    @Resource
    private DriverSdkService driverSdkService;

    @Resource
    private BatchService batchService;

    @Resource
    private RedisUtil redisUtil;
    // 消费监听
    @KafkaListener(topics = {Common.KAFKA.TOPIC_EXCHANGE_DRIVER_EVENT},groupId = "${driverEventGroup}")
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("DriverEventCon简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
        String jsonstr = record.value().toString();
        DriverEvent driverEvent= JsonPaseUtil.JSONStrToObj(jsonstr, DriverEvent.class);

        //处理完下命令给DriverMEtaConsumer，由DriverMetaConsumer进行统一处理
        switch (driverEvent.getType()) {
            //第一次握手
            case Common.Driver.Event.DRIVER_HANDSHAKE:
                DriverConfiguration driverConfiguration = new DriverConfiguration(
                        Common.Driver.Type.DRIVER,
                        Common.Driver.Event.DRIVER_HANDSHAKE_BACK,
                        null,
                        Common.Response.OK,
                        driverEvent.getServiceName()
                );
                kafkaProducer.send(
                        Common.KAFKA.TOPIC_EXCHANGE_METADATA,
                        JsonPaseUtil.ObjToJsonStr(driverConfiguration));
                break;
            case Common.Driver.Event.DRIVER_REGISTER:
                driverConfiguration = new DriverConfiguration(
                        Common.Driver.Type.DRIVER,
                        Common.Driver.Event.DRIVER_REGISTER_BACK,
                        null,
                        Common.Response.OK,
                        driverEvent.getServiceName() //kafka没有单独通道区分，通过serviceName来进行区分
                );
                try {
                    //写入（或者更新）数据库的动作，涉及到 driver  driver-attribute  point-attribute三张表
                    driverSdkService.driverRegister(Convert.convert(DriverRegister.class, driverEvent.getContent()));
                } catch (Exception e) {
                    driverConfiguration.setResponse(e.getMessage());
                }
                kafkaProducer.send(
                        Common.KAFKA.TOPIC_EXCHANGE_METADATA,
                        JsonPaseUtil.ObjToJsonStr(driverConfiguration));
                break;
            case Common.Driver.Event.DRIVER_METADATA_SYNC:
                driverConfiguration = new DriverConfiguration(
                        Common.Driver.Type.DRIVER,
                        Common.Driver.Event.DRIVER_METADATA_SYNC_BACK,
                        null,
                        Common.Response.OK,
                        driverEvent.getServiceName()
                );
                try {
                    driverConfiguration.setContent(batchService.batchDriverMetadata(driverEvent.getServiceName()));
                } catch (Exception e) {
                    driverConfiguration.setResponse(e.getMessage());
                }
                kafkaProducer.send(
                        Common.KAFKA.TOPIC_EXCHANGE_METADATA,
                        JsonPaseUtil.ObjToJsonStr(driverConfiguration));
                break;
            case Common.Driver.Event.HEARTBEAT:
                redisUtil.setKey(
                        Common.Cache.DRIVER_STATUS_KEY_PREFIX + driverEvent.getServiceName(),
                        driverEvent.getContent(),
                        driverEvent.getTimeOut(),
                        driverEvent.getTimeUnit()
                );
                break;
            case Common.Driver.Event.ERROR:
               /* //TODO 去重
                threadPoolExecutor.execute(() -> eventService.addDriverEvent(driverEvent));*/
            default:
                //log.error("Invalid event type, {}", driverEvent.getType());
                break;
        }
        ack.acknowledge();
    }

}
